package com.csx.governance.traffic.mq.rocketmq.config;

import com.csx.governance.traffic.core.TrafficGovernanceRuleLocator;
import com.csx.governance.traffic.core.metadata.ServiceMetadataLocator;
import com.csx.governance.traffic.mq.rocketmq.*;
import com.csx.governance.traffic.mq.rocketmq.hook.TrafficConsumeMessageHook;
import com.csx.governance.traffic.mq.rocketmq.hook.TrafficProduceMessageHook;
import com.csx.governance.traffic.mq.rocketmq.model.RocketMqServiceRule;
import com.csx.governance.traffic.route.swimlane.SwimLaneGovernanceRuleLocator;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.hook.SendMessageHook;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.core.RocketMQReplyListener;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
import org.springframework.core.env.Environment;

/**
 * @author cuisongxu
 * @date 2023/2/28 16:50
 */
@ConditionalOnProperty(value = "traffic.governance.mq.enabled", havingValue = "true", matchIfMissing = true)
@ConditionalOnClass(RocketMQListener.class)
@Configuration
@EnableConfigurationProperties(RocketMqRulesConfiguration.class)
@EnableAspectJAutoProxy
public class TrafficGovernanceRocketMQAutoConfiguration {

    @Bean
    public TrafficProduceMessageHook trafficProduceMessageHook(TrafficGovernanceRuleLocator<RocketMqServiceRule> trafficGovernanceRuleLocator,
                                                               ApplicationContext applicationContext) {
        return new TrafficProduceMessageHook(trafficGovernanceRuleLocator, applicationContext);
    }

    @Bean
    public TrafficConsumeMessageHook trafficConsumeMessageHook(SwimLaneGovernanceRuleLocator swimLaneGovernanceRuleLocator,
                                                               Environment environment,
                                                               ServiceMetadataLocator serviceMetadataLocator) {
        String localApplicationName = environment.getProperty("spring.application.name");
        return new TrafficConsumeMessageHook(swimLaneGovernanceRuleLocator, localApplicationName, serviceMetadataLocator);
    }

    @Bean
    public BeanPostProcessor trafficMqConsumerBeanPostProcessor(ApplicationContext applicationContext,
                                                                TrafficGovernanceRuleLocator<RocketMqServiceRule> rocketMqServiceRuleTrafficGovernanceRuleLocator,
                                                                ServiceMetadataLocator serviceMetadataLocator) {
        return new TrafficMQConsumerBeanPostProcessor(applicationContext, rocketMqServiceRuleTrafficGovernanceRuleLocator, serviceMetadataLocator);
    }

    @Bean
    public BeanPostProcessor trafficMqProducerBeanPostProcessor(ApplicationContext applicationContext) {
        return new TrafficMQProducerBeanPostProcessor(applicationContext);
    }

    @Bean
    public MessageListener trafficConsumeMessageListenerOrderly() {
        return new TrafficConsumeMessageListenerOrderly();
    }

    @Bean
    public MessageListener trafficConsumeMessageListenerConcurrently() {
        return new TrafficConsumeMessageListenerConcurrently();
    }

    @Bean
    public MessageQueueSelector trafficMessageQueueSelector() {
        return new TrafficMessageQueueSelector();
    }

    @Bean
    public RocketMQListener<String> trafficRocketMQListener() {
        return new TrafficRocketMQListener();
    }

    @Bean
    public RocketMQPushConsumerLifecycleListener trafficRocketMQPushConsumerLifecycleListener() {
        return new TrafficRocketMQPushConsumerLifecycleListener();
    }

    @Bean
    public RocketMQReplyListener<String, String> trafficRocketMQReplyListener() {
        return new TrafficRocketMQReplyListener();
    }

    @Bean
    public TrafficGovernanceRuleLocator<RocketMqServiceRule> rocketMqServiceRuleTrafficGovernanceRuleLocator(RocketMqRulesConfiguration rules) {
        return new RocketMqGovernanceRuleLocator(rules);
    }

    @Bean
    public RocketMqConsumeAspect rocketMqConsumeAspect() {
        return new RocketMqConsumeAspect();
    }
}
